package com.kedacom.demo.websocket.rpc;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import javax.websocket.ClientEndpoint;
import javax.websocket.ClientEndpointConfig;
import javax.websocket.CloseReason;
import javax.websocket.ContainerProvider;
import javax.websocket.DeploymentException;
import javax.websocket.HandshakeResponse;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.WebSocketContainer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.kedacom.demo.utils.JacksonUtils;

@ClientEndpoint(configurator = AbstractRPCWebSocketClient.ClientConfigurator.class)
public abstract class AbstractRPCWebSocketClient {

	protected Logger logger = LoggerFactory.getLogger(getClass());

	protected static PendingRequestHolder pendingRequestHolder = PendingRequestHolder.getInstance();

	protected String wsServerUrl;
	protected int timeout = 15;
	protected boolean autoReconnect = true; // 断链后是否自动重连
	protected int reconnectInterval = 3000; // 断链重连间隔（毫秒）
	protected Session session;

	public AbstractRPCWebSocketClient() {
	}

	public AbstractRPCWebSocketClient(String wsServerUrl) {
		this.wsServerUrl = wsServerUrl;
	}

	public void setWsServerUrl(String wsServerUrl) {
		this.wsServerUrl = wsServerUrl;
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	public void setAutoReconnect(boolean autoReconnect) {
		this.autoReconnect = autoReconnect;
	}

	public void setReconnectInterval(int reconnectInterval) {
		this.reconnectInterval = reconnectInterval;
	}

	/**
	 * 连接WebSocket Server
	 */
	public boolean connect() {
		if (wsServerUrl == null || "".equals(wsServerUrl)) {
			logger.error("wsServerUrl不能为空");
			return false;
		}

		try {
			WebSocketContainer container = ContainerProvider.getWebSocketContainer();
			session = container.connectToServer(this, new URI(wsServerUrl));
			return true;
		} catch (DeploymentException e) {
			logger.error("DeploymentException: {}", e.getMessage());
		} catch (IOException e) {
			logger.error("IOException: {}", e.getMessage());
		} catch (URISyntaxException e) {
			logger.error("URISyntaxException: {}", e.getMessage());
		}
		return false;
	}

	@OnOpen
	public void onOpen(Session session) {
		this.session = session;
		logger.info("与服务器建链成功");
	}

	@OnMessage
	public void onMessage(String message) {
		logger.debug("收到消息：{}", message);
		Response response = JacksonUtils.fromJsonStringSilently(message, Response.class);
		String requestId = response.getId();
		if (requestId != null && !"".equals(requestId)) {
			PendingRequest pendingRequest = pendingRequestHolder.get(requestId);
			if (pendingRequest != null) {
				pendingRequest.getQueue().add(message);
			} else {
				logger.error("找不到对应的请求，requestId={}", requestId);
			}
		} else {
			processPushMessage(message);
		}
	}

	/** 处理推送消息 */
	public abstract void processPushMessage(String message);

	@OnClose
	public void onClose(CloseReason reason) {
		logger.info("连接关闭: {}", JacksonUtils.toJsonString(reason));
		session = null;
		reconnect();
	}

	@OnError
	public void onError(Throwable throwable) {
		logger.error("客户端出错：{}", throwable.getMessage());
	}

	/**
	 * 发送RPC请求
	 */
	public <T extends Response> T request(String requestType, String requestData, Class<T> reponseClass)
			throws ConnectionClosedException, ConnectionErrorException {
		return request(new Request(requestType, requestData), reponseClass);
	}

	/**
	 * 发送RPC请求
	 */
	public <T extends Response> T request(Request request, Class<T> reponseClass)
			throws ConnectionClosedException, ConnectionErrorException {
		if (session == null) {
			logger.error("连接已关闭，暂时无法发送请求");
			throw new ConnectionClosedException();
		}

		PendingRequest pendingRequest = new PendingRequest(request.getId());
		pendingRequestHolder.put(request.getId(), pendingRequest);
		try {
			sendMessage(JacksonUtils.toJsonString(request));
			String response = this.waitForResponse(pendingRequest);
			return JacksonUtils.fromJsonStringSilently(response, reponseClass);
		} catch (InterruptedException | TimeoutException | IOException e) {
			throw new ConnectionErrorException(e);
		} finally {
			// remove the pending request from pending request holder
			pendingRequestHolder.remove(pendingRequest.getRequestId());
		}
	}

	public void sendMessage(String message) throws IOException {
		session.getBasicRemote().sendText(message);
	}

	/**
	 * 等待服务器响应请求
	 */
	private String waitForResponse(PendingRequest pendingRequest) throws InterruptedException, TimeoutException {
		LinkedBlockingQueue<String> replyHandoff = pendingRequest.getQueue();
		// receive the response
		String response = (timeout < 0) ? replyHandoff.take() : replyHandoff.poll(timeout, TimeUnit.SECONDS);
		if (response == null) {
			// timeout
			throw new TimeoutException(
					"Timeout while waiting for response of request with id: " + pendingRequest.getRequestId());
		}

		logger.debug("receive response '{}' for request with id: ", response, pendingRequest.getRequestId());
		return response;
	}

	private void reconnect() {
		while (session == null) {
			if (!autoReconnect) {
				return;
			}
			try {
				logger.info("尝试重连服务器...");
				this.connect();
				Thread.sleep(reconnectInterval);
			} catch (Exception e) {
				logger.error("reconnect error {}", e.getMessage());
			}
		}
	}

	public void destroy() {
		autoReconnect = false;
		if (session == null) {
			return;
		}
		try {
			session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "客户端主动断开连接。"));
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static class ClientConfigurator extends ClientEndpointConfig.Configurator {

		@Override
		public void beforeRequest(Map<String, List<String>> headers) {
			super.beforeRequest(headers);
		}

		@Override
		public void afterResponse(HandshakeResponse hr) {
			super.afterResponse(hr);
		}
	}

}
